package com.meiya.flink.schema;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.meiya.flink.config.AppConfig;

/**
 * 表定义类，包含所有表的创建SQL
 */
public class TableDefinitions {
    private static final Logger logger = LoggerFactory.getLogger(TableDefinitions.class);

    /**
     * 获取创建Kafka账户源表的SQL
     */
    public static String getCreateKafkaAccountSourceTableSQL() {
        return "CREATE TABLE IF NOT EXISTS account_source (\n" +
                "  clue_id VARCHAR,\n" +
                "  zhkhmc STRING,\n" +
                "  khrzjhm STRING,\n" +
                "  jykh STRING,\n" +
                "  jyzh STRING,\n" +
                "  zhkhsj STRING,\n" +
                "  zhye STRING,\n" +
                "  kyye STRING,\n" +
                "  bz STRING,\n" +
                "  khwddm STRING,\n" +
                "  khwd STRING,\n" +
                "  zhzt STRING,\n" +
                "  chbzmc STRING,\n" +
                "  khrzjlx STRING,\n" +
                "  xhrq STRING,\n" +
                "  zhlx STRING,\n" +
                "  khlxfs STRING,\n" +
                "  txdz STRING,\n" +
                "  lxdh STRING,\n" +
                "  dlr STRING,\n" +
                "  dlrdh STRING,\n" +
                "  bzbz STRING,\n" +
                "  khsf STRING,\n" +
                "  khcs STRING,\n" +
                "  zhkhyh STRING,\n" +
                "  khdm STRING,\n" +
                "  frdb STRING,\n" +
                "  khgszzhm STRING,\n" +
                "  frdbzjhm STRING,\n" +
                "  zzdz STRING,\n" +
                "  yzbm STRING,\n" +
                "  dbrzjhm STRING,\n" +
                "  yxdz STRING,\n" +
                "  glzjzh STRING,\n" +
                "  dsnsh STRING,\n" +
                "  dwdh STRING,\n" +
                "  dbrzjlx STRING,\n" +
                "  zzdh STRING,\n" +
                "  frdbzjlx STRING,\n" +
                "  gsnsh STRING,\n" +
                "  dwdz STRING,\n" +
                "  gzdw STRING,\n" +
                "  xhwd STRING,\n" +
                "  zhjysj STRING,\n" +
                "  zhxhyh STRING,\n" +
                "  rwlsh STRING,\n" +
                "  a_yhlx STRING,\n" +
                "  origin_data_id BIGINT,\n" +
                "  crrq STRING,\n" +
                "  ukey STRING,\n" +
                "  task_record_id STRING,\n" +
                "  processing_time AS PROCTIME(),\n" +
                "  event_time AS TO_TIMESTAMP(crrq, 'yyyy-MM-dd HH:mm:ss'),\n" +
                "  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + AppConfig.getKafkaAccountTopic() + "',\n" +
                "  'properties.bootstrap.servers' = '" + AppConfig.getKafkaBrokers() + "',\n" +
                "  'properties.group.id' = '" + AppConfig.getKafkaConsumerGroupId() + "',\n" +
                // todo 设置消费者的起始位置为最新 latest-offset
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json',\n" +
                "  'json.fail-on-missing-field' = 'false',\n" +
                "  'json.ignore-parse-errors' = 'true',\n" +
                // todo 设置 auto.offset.reset latest
                "  'properties.auto.offset.reset' = 'latest',\n" +
                "  'properties.enable.auto.commit' = 'true',\n" +
                "  'properties.auto.commit.interval.ms' = '1000',\n" +
                "  'properties.session.timeout.ms' = '30000',\n" +
                "  'properties.request.timeout.ms' = '40000',\n" +
                "  'properties.max.poll.records' = '500'\n" +
                ")";
    }

    /**
     * 获取创建Kafka流水源表的SQL
     */
    public static String getCreateKafkaSourceTableSQL() {
        return "CREATE TABLE IF NOT EXISTS transaction_source (\n" +
                "  clue_id VARCHAR,\n" +
                "  jykh STRING,\n" +
                "  jyzh STRING,\n" +
                "  jyfhm STRING,\n" +
                "  jyfzjhm STRING,\n" +
                "  jysj TIMESTAMP(3),\n" +
                "  jyje DECIMAL(20, 2),\n" +
                "  jyye DECIMAL(20, 2),\n" +
                "  sfbz STRING,\n" +
                "  jydszkh STRING,\n" +
                "  xjbz STRING,\n" +
                "  dshm STRING,\n" +
                "  dssfzh STRING,\n" +
                "  dskhyh STRING,\n" +
                "  zysm STRING,\n" +
                "  jybz STRING,\n" +
                "  jywdmc STRING,\n" +
                "  jyfsd STRING,\n" +
                "  jysfcg STRING,\n" +
                "  cph STRING,\n" +
                "  ipdz STRING,\n" +
                "  macdz STRING,\n" +
                "  dsjyye DECIMAL(20, 2),\n" +
                "  jylsh STRING,\n" +
                "  rzh STRING,\n" +
                "  pzzl STRING,\n" +
                "  pzh STRING,\n" +
                "  jygyh STRING,\n" +
                "  bz STRING,\n" +
                "  shmc STRING,\n" +
                "  jylx STRING,\n" +
                "  cxfkjgyy STRING,\n" +
                "  task_record_id STRING,\n" +
                "  a_yhlx STRING,\n" +
                "  action_type INT,\n" +
                "  from_jykh STRING,\n" +
                "  to_jydszkh STRING,\n" +
                "  crrq STRING,\n" +
                "  ukey STRING,\n" +
                "  processing_time AS PROCTIME(),\n" +
                "  event_time AS TO_TIMESTAMP(crrq, 'yyyy-MM-dd HH:mm:ss'),\n" +
                "  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + AppConfig.getKafkaTransactionTopic() + "',\n" +
                "  'properties.bootstrap.servers' = '" + AppConfig.getKafkaBrokers() + "',\n" +
                "  'properties.group.id' = '" + AppConfig.getKafkaConsumerGroupId() + "',\n" +
                // todo 设置消费者的起始位置为最新 latest-offset
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json',\n" +
                "  'json.fail-on-missing-field' = 'false',\n" +
                "  'json.ignore-parse-errors' = 'true',\n" +
                // todo auto.offset.reset latest
                "  'properties.auto.offset.reset' = 'latest',\n" +
                "  'properties.enable.auto.commit' = 'true',\n" +
                "  'properties.auto.commit.interval.ms' = '1000',\n" +
                "  'properties.session.timeout.ms' = '30000',\n" +
                "  'properties.request.timeout.ms' = '40000',\n" +
                "  'properties.max.poll.records' = '500'\n" +
                ")";
    }

    /**
     * 获取创建Doris交易明细表的SQL
     */
    public static String getCreateDorisTransactionTableSQL() {
        return "CREATE TABLE IF NOT EXISTS flink_sr_transcation_detail (\n" +
                "  action_type INT,\n" +
                "  clue_id VARCHAR,\n" +
                "  sfbz STRING,\n" +
                "  jybz STRING,\n" +
                "  jykh STRING,\n" +
                "  jyzh STRING,\n" +
                "  jyje DECIMAL(20, 2),\n" +
                "  jyye DECIMAL(20, 2),\n" +
                "  jysj STRING,\n" +
                "  dshm STRING,\n" +
                "  jydszkh STRING,\n" +
                "  jykhh STRING,\n" +
                "  jyfhm STRING,\n" +
                "  jyfzjhm STRING,\n" +
                "  ssfkyhjgdm STRING,\n" +
                "  mxlx STRING,\n" +
                "  ksx STRING,\n" +
                "  xjbz STRING,\n" +
                "  dssfzh STRING,\n" +
                "  dskhyh STRING,\n" +
                "  zysm STRING,\n" +
                "  qd STRING,\n" +
                "  jyyhmc STRING,\n" +
                "  jygj STRING,\n" +
                "  jyhwdd STRING,\n" +
                "  jywdmc STRING,\n" +
                "  jyfsd STRING,\n" +
                "  jycs STRING,\n" +
                "  jysfcg STRING,\n" +
                "  cph STRING,\n" +
                "  ipdz STRING,\n" +
                "  macdz STRING,\n" +
                "  dsjyye DECIMAL(20, 2),\n" +
                "  jylsh STRING,\n" +
                "  rzh STRING,\n" +
                "  pzzl STRING,\n" +
                "  pzh STRING,\n" +
                "  jygyh STRING,\n" +
                "  bz STRING,\n" +
                "  shh STRING,\n" +
                "  shmc STRING,\n" +
                "  jylx STRING,\n" +
                "  cxfkjgyy STRING,\n" +
                "  ybje DECIMAL(20, 2),\n" +
                "  mbdm STRING,\n" +
                "  mbmc STRING,\n" +
                "  myje STRING,\n" +
                "  fkflx STRING,\n" +
                "  skflx STRING,\n" +
                "  dbrmc STRING,\n" +
                "  dbrzjlx STRING,\n" +
                "  dbrzjhm STRING,\n" +
                "  fkfzhlx STRING,\n" +
                "  fkfzjlx STRING,\n" +
                "  fkfgj STRING,\n" +
                "  fkfyhszd STRING,\n" +
                "  sffzhlx STRING,\n" +
                "  skfzjlx STRING,\n" +
                "  skfgj STRING,\n" +
                "  skfyhszd STRING,\n" +
                "  rmbje DECIMAL(20, 2),\n" +
                "  sjlyid INT,\n" +
                "  sjlylx STRING,\n" +
                "  a_yhlx STRING,\n" +
                "  from_jykh STRING,\n" +
                "  to_jydszkh STRING,\n" +
                "  ukey STRING,\n" +
                "  task_record_id STRING,\n" +
                "  del STRING,\n" +
                "  source STRING,\n" +
                "  crrq STRING,\n" +
                "  origin_data_id BIGINT,\n" +
                "  event_time AS TO_TIMESTAMP(crrq, 'yyyy-MM-dd HH:mm:ss'),\n" +
                "  processing_time AS PROCTIME(),\n" +
                "  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'doris',\n" +
                "  'fenodes' = '" + AppConfig.getDorisHttpNodes() + "',\n" +
                "  'table.identifier' = '" + AppConfig.getDorisDatabase() + ".sr_transcation_detail',\n" +
                "  'username' = '" + AppConfig.getDorisUsername() + "',\n" +
                "  'password' = '" + AppConfig.getDorisPassword() + "',\n" +
                "  'sink.properties.format' = 'json',\n" +
                "  'sink.properties.read_json_by_line' = 'true',\n" +
                "  'sink.buffer-flush.interval' = '5s',\n" +
                "  'sink.max-retries' = '5',\n" +
                "  'doris.request.connect.timeout' = '180000',\n" +
                "  'doris.request.read.timeout' = '180000',\n" +
                "  'doris.request.query.timeout' = '360000',\n" +
                "  'doris.request.retries' = '5',\n" +
                "  'doris.request.tablet.size' = '5',\n" +
                "  'doris.batch.size' = '2048',\n" +
                "  'doris.exec.mem.limit' = '8589934592'\n" +
                ")";
    }

    /**
     * 获取创建Doris账户表的SQL
     */
    public static String getCreateDorisAccountTableSQL() {
        return "CREATE TABLE IF NOT EXISTS flink_sr_account (\n" +
                "  clue_id VARCHAR,\n" +
                "  zhkhmc STRING,\n" +
                "  khrzjhm STRING,\n" +
                "  jykh STRING,\n" +
                "  jyzh STRING,\n" +
                "  zhkhsj TIMESTAMP(3),\n" +
                "  zhye DECIMAL(20, 2),\n" +
                "  kyye DECIMAL(20, 2),\n" +
                "  bz STRING,\n" +
                "  khwddm STRING,\n" +
                "  khwd STRING,\n" +
                "  zhzt STRING,\n" +
                "  chbzmc STRING,\n" +
                "  khrzjlx STRING,\n" +
                "  xhrq STRING,\n" +
                "  zhlx STRING,\n" +
                "  khlxfs STRING,\n" +
                "  txdz STRING,\n" +
                "  lxdh STRING,\n" +
                "  dlr STRING,\n" +
                "  dlrdh STRING,\n" +
                "  bzbz STRING,\n" +
                "  khsf STRING,\n" +
                "  khcs STRING,\n" +
                "  zhkhyh STRING,\n" +
                "  khdm STRING,\n" +
                "  frdb STRING,\n" +
                "  khgszzhm STRING,\n" +
                "  frdbzjhm STRING,\n" +
                "  zzdz STRING,\n" +
                "  yzbm STRING,\n" +
                "  dbrzjhm STRING,\n" +
                "  yxdz STRING,\n" +
                "  glzjzh STRING,\n" +
                "  dsnsh STRING,\n" +
                "  dwdh STRING,\n" +
                "  dbrzjlx STRING,\n" +
                "  zzdh STRING,\n" +
                "  frdbzjlx STRING,\n" +
                "  gsnsh STRING,\n" +
                "  dwdz STRING,\n" +
                "  gzdw STRING,\n" +
                "  xhwd STRING,\n" +
                "  zhjysj TIMESTAMP(3),\n" +
                "  zhxhyh STRING,\n" +
                "  rwlsh STRING,\n" +
                "  a_yhlx STRING,\n" +
                "  origin_data_id STRING,\n" +
                "  crrq STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'doris',\n" +
                "  'fenodes' = '" + AppConfig.getDorisHttpNodes() + "',\n" +
                "  'table.identifier' = '" + AppConfig.getDorisDatabase() + ".sr_account',\n" +
                "  'username' = '" + AppConfig.getDorisUsername() + "',\n" +
                "  'password' = '" + AppConfig.getDorisPassword() + "',\n" +
                "  'sink.properties.format' = 'json',\n" +
                "  'sink.properties.read_json_by_line' = 'true',\n" +
                "  'doris.request.connect.timeout' = '180000',\n" +
                "  'doris.request.read.timeout' = '180000',\n" +
                "  'doris.request.query.timeout' = '360000',\n" +
                "  'doris.request.retries' = '5',\n" +
                "  'doris.request.tablet.size' = '5',\n" +
                "  'doris.batch.size' = '2048',\n" +
                "  'doris.exec.mem.limit' = '8589934592',\n" +
                "  'sink.max-retries' = '5'\n" +
                ")";
    }

    /**
     * 获取创建标签结果表的SQL
     */
    public static String getCreateTagsResultTableSQL() {
        return "CREATE TABLE IF NOT EXISTS tags_result (\n" +
                "  mid BIGINT,\n" +
                "  clue_id VARCHAR,\n" +
                "  tags STRING,\n" +
                "  jykh STRING,\n" +
                "  jyfhm STRING,\n" +
                "  khzjhm STRING,\n" +
                "  khyhmc STRING,\n" +
                "  zjyje DECIMAL(20, 2),\n" +
                "  jycz BIGINT,\n" +
                "  cjsj TIMESTAMP(3),\n" +
                "  xgsj TIMESTAMP(3),\n" +
                "  ukeys STRING,\n" +
                "  PRIMARY KEY (mid, clue_id, jykh) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '" + AppConfig.getKafkaTagsResultTopic() + "',\n" +
                "  'properties.bootstrap.servers' = '" + AppConfig.getKafkaBrokers() + "',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json',\n" +
                "  'value.json.fail-on-missing-field' = 'false',\n" +
                "  'value.json.ignore-parse-errors' = 'true',\n" +
                // 设置 max.request.size 为 10MB，避免发送失败
                "  'properties.max.request.size' = '10485760',\n" +
                "  'properties.buffer.memory' = '67108864',\n" +
                "  'properties.batch.size' = '131072',\n" +
                "  'properties.linger.ms' = '10',\n" +
                "  'properties.compression.type' = 'snappy'\n" +
                ")";
    }

    /**
     * 获取创建已处理交易表的SQL
     */
    public static String getCreateProcessedTradesTableSQL() {
        return "CREATE TABLE IF NOT EXISTS processed_trades (\n" +
                "  clue_id VARCHAR,\n" +
                "  jykh STRING,\n" +
                "  mid STRING,\n" +
                "  last_process_time TIMESTAMP(3),\n" +
                "  PRIMARY KEY (clue_id, jykh) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'doris',\n" +
                "  'fenodes' = '" + AppConfig.getDorisHttpNodes() + "',\n" +
                "  'table.identifier' = '" + AppConfig.getDorisDatabase() + ".processed_trades',\n" +
                "  'username' = '" + AppConfig.getDorisUsername() + "',\n" +
                "  'password' = '" + AppConfig.getDorisPassword() + "',\n" +
                "  'sink.properties.format' = 'json',\n" +
                "  'sink.properties.read_json_by_line' = 'true',\n" +
                "  'doris.request.connect.timeout' = '180000',\n" +
                "  'doris.request.read.timeout' = '180000',\n" +
                "  'doris.request.query.timeout' = '360000',\n" +
                "  'doris.request.retries' = '5',\n" +
                "  'doris.request.tablet.size' = '5',\n" +
                "  'doris.batch.size' = '2048',\n" +
                "  'doris.exec.mem.limit' = '8589934592',\n" +
                "  'sink.max-retries' = '5'\n" +
                ")";
    }

    /**
     * 获取创建Doris标签结果表的SQL
     */
    public static String getCreateDorisTagsResultTableSQL() {
        return "CREATE TABLE IF NOT EXISTS doris_tags_result (\n" +
                "  mid BIGINT,\n" +
                "  clue_id VARCHAR,\n" +
                "  tags STRING,\n" +
                "  jykh STRING,\n" +
                "  jyfhm STRING,\n" +
                "  khzjhm STRING,\n" +
                "  khyhmc STRING,\n" +
                "  zjyje DECIMAL(20, 2),\n" +
                "  jycz BIGINT,\n" +
                "  cjsj TIMESTAMP(3),\n" +
                "  xgsj TIMESTAMP(3),\n" +
                "  ukeys STRING,\n" +
                "  PRIMARY KEY (mid, clue_id, jykh) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'doris',\n" +
                "  'fenodes' = '" + AppConfig.getDorisHttpNodes() + "',\n" +
                "  'table.identifier' = '" + AppConfig.getDorisDatabase() + ".processed_tags_result',\n" +
                "  'username' = '" + AppConfig.getDorisUsername() + "',\n" +
                "  'password' = '" + AppConfig.getDorisPassword() + "',\n" +
                "  'sink.properties.format' = 'json',\n" +
                "  'sink.properties.read_json_by_line' = 'true',\n" +
                "  'sink.buffer-flush.interval' = '5s',\n" +
                "  'sink.max-retries' = '3',\n" +
                "  'doris.request.connect.timeout' = '180000',\n" +
                "  'doris.request.read.timeout' = '180000',\n" +
                "  'doris.request.query.timeout' = '360000',\n" +
                "  'doris.request.retries' = '5',\n" +
                "  'doris.request.tablet.size' = '5',\n" +
                "  'doris.batch.size' = '2048',\n" +
                "  'doris.exec.mem.limit' = '8589934592',\n" +
                "  'sink.max-retries' = '5'\n" +
                ")";
    }

    /**
     * 获取创建已处理标签表的SQL
     */
    public static String getCreateProcessedTagsTableSQL() {
        return "CREATE TABLE IF NOT EXISTS processed_tags (\n" +
                "  clue_id VARCHAR,\n" +
                "  jykh STRING,\n" +
                "  mid STRING,\n" +
                "  last_process_time TIMESTAMP(3),\n" +
                "  PRIMARY KEY (clue_id, jykh, mid) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'doris',\n" +
                "  'fenodes' = '" + AppConfig.getDorisHttpNodes() + "',\n" +
                "  'table.identifier' = '" + AppConfig.getDorisDatabase() + ".processed_tags',\n" +
                "  'username' = '" + AppConfig.getDorisUsername() + "',\n" +
                "  'password' = '" + AppConfig.getDorisPassword() + "',\n" +
                "  'sink.properties.format' = 'json',\n" +
                "  'sink.properties.read_json_by_line' = 'true',\n" +
                "  'doris.request.connect.timeout' = '180000',\n" +
                "  'doris.request.read.timeout' = '180000',\n" +
                "  'doris.request.query.timeout' = '360000',\n" +
                "  'doris.request.retries' = '5',\n" +
                "  'doris.request.tablet.size' = '5',\n" +
                "  'doris.batch.size' = '2048',\n" +
                "  'doris.exec.mem.limit' = '8589934592',\n" +
                "  'sink.max-retries' = '5'\n" +
                ")";
    }

}